/*
 *
 * This program is free software; you can redistribute it and/or modify 
 * it under the terms of the GNU General Public License version 2 as 
 * published by the Free Software Foundation.
 *
 * thread_pool.cpp is for what ...
 *
 * Version: ***: thread_pool.cpp  Sat Jul  2 10:07:35 2011 poponion Exp $
 *
 * Authors:
 *   Author poponion <huangcm85@gmail.com>
 *     -some work detail if you want 
 *
 */

#include "thread_pool.h"

int ThreadPool::init(int thread)
{
  int ret = 0;
  thread_num_ = 0;
  pthread_mutex_init(&mutex_, NULL);
  pthread_cond_init(&cond_, NULL);
  init_ = true;
  threads_ = reinterpret_cast<pthread_t *>(malloc(sizeof(pthread_t) * thread));
  if (NULL == threads_)
  {
    fprintf(stderr, "malloc memory failed\n");
    ret = -1;
  }
  else
  {
    for (int index = 0; index < thread; ++index)
    {
      ret = pthread_create(&threads_[index], NULL, thread_run, this);
      if (0 != ret) 
      {
        fprintf(stderr, "create thread failed\n");
        break;
      }
      else
      {
        ++thread_num_;
      }
    }
  }
  return ret;
}

void * ThreadPool::thread_run(void *arg)
{
  ThreadPool * tpool = reinterpret_cast<ThreadPool *>(arg);
  tpool->run();
}

void * ThreadPool::run()
{
  fprintf(stderr, "Thread 0x%x start working\n", pthread_self());
  while(true)
  {
    pthread_mutex_lock(&mutex_);
    while(0 == tasks_.length() && init_)
    {
      pthread_cond_wait(&cond_, &mutex_);
    }
    if (!init_)
    {
      fprintf(stderr, "Thread pool not init, thread 0x%x exit\n", pthread_self());
      pthread_mutex_unlock(&mutex_);
      return (void *)0;
    }
    
    DefaultTask task;
    if (0 != tasks_.length())
    {
      task = tasks_.pop()->data_;
    }
    pthread_mutex_unlock(&mutex_);
    fprintf(stderr, "Thread 0x%x handle task %d\n", pthread_self(), task.get_magic());
    task.run();
  }
}

int ThreadPool::add_task(DefaultTask & task)
{
  int ret = 0;
  pthread_mutex_lock(&mutex_);
  tasks_.push(task);
  pthread_mutex_unlock(&mutex_);
  pthread_cond_signal(&cond_);
  return ret;
}

void ThreadPool::destroy()
{
  //join all thread and free memory
  if (!init_)
  {
    fprintf(stderr, "ThreadPool not init yet\n");
  }
  else
  {
    init_ = false;
    pthread_cond_broadcast(&cond_);
    for (int index = 0; index < thread_num_; ++index)
    {
      pthread_join(threads_[index], NULL);
      //fprintf(stderr, "join thread 0x%x\n", threads_[index]);
    }
    pthread_mutex_destroy(&mutex_);
    pthread_cond_destroy(&cond_);
    //free(threads_);
    threads_ = NULL;
  }
}



int main(int argc, char *argv[])
{
  ThreadPool tp;
  tp.init(20);
 
  for(int idx = 0; idx < 1000; ++idx)
  {
    DefaultTask task;
    task.set_magic(idx);
    tp.add_task(task);
  }
  

  for(int idx = 0; idx < 200; ++idx)
  {
    DefaultTask task;
    task.set_magic(idx + 100);
    tp.add_task(task);
  }
  tp.destroy();
}
